package test.top2.com.atguigu.app.fun;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import test.top2.com.atguigu.common.GmallConfig;
import test.top2.com.atguigu.utils.DruidDSUtil;

import java.sql.PreparedStatement;
import java.util.Collection;
import java.util.Map;
import java.util.Set;

/**
 * ClassName: DIMSinkFunction
 * Package: test.top2.com.atguigu.app.fun
 * Description:
 *
 * @Author ChenJun(有志男青年)
 * @Create 2023/4/27 20:17
 * @Version 1.0
 */
public class DIMSinkFunction extends RichSinkFunction<JSONObject> {

    private DruidDataSource druidDataSource;


    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource = DruidDSUtil.getDruidDataSource();
    }


    //value:{"database":"gmall-221109-flink","table":"comment_info","type":"insert","ts":1669162958,"xid":1111,"xoffset":13941,"data":{"id":1595211185799847960,"user_id":119,"nick_name":null,"head_img":null,"sku_id":31,"spu_id":10,"order_id":987,"appraise":"1204","comment_txt":"评论内容：48384811984748167197482849234338563286217912223261","create_time":"2022-08-02 08:22:38","operate_time":null},"sink_table":"dim_comment_info"}
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {

        //获取连接
        DruidPooledConnection connection = druidDataSource.getConnection();

        //获取sql     upsert into db.tn(id,name,sex) values('1001','zs','male')
        String upsertSql = genSql(value.getString("sink_table"),value.getJSONObject("data"));

        //预编译sql
        PreparedStatement preparedStatement = connection.prepareStatement(upsertSql);

        //执行sql
         preparedStatement.execute();

         //提交
        connection.commit();

        //资源关闭
        preparedStatement.close();
        //归还
        connection.close();

    }

    private String genSql(String sink_table, JSONObject data) {
        //获取字段
        Set<String> strings = data.keySet();

        //获取列值
        Collection<Object> values = data.values();

        return "upsert into "+ GmallConfig.HBASE_SCHEMA+"."+sink_table+"("+
                StringUtils.join(strings,",")+") values ('"+
                StringUtils.join(values,"','")+")";
    }

    @Override
    public void close() throws Exception {
        druidDataSource.close();
    }
}
